package com.youxin.dataStream.partition;

import com.youxin.dataStream.custormSource.ParalleSource;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 自定义的分区和flink的默认的分区实现是不一样的
 * //随机分区
 * mapTuple.shuffle();
 * //再均衡，消除数据倾斜,其实跟上面的差不多
 * mapTuple.rebalance();
 *
 * // 上游的数据单个分给多个节点，不会全量分区
 * 上游：a b 下游的 1 2 3 4 则   a--> 1,2   b---> 3,4
 * mapTuple.rescale();
 *
 * //广播分区
 * mapTuple.broadcast();
 *
 * 注：这里面没有rangPartition ,sortpartition之类的? 这是什么原因呢？可能是数据都是单条的，不需要这样的操作
 *
 */
public class StreamWithPartition {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);
        DataStreamSource<Long> source = env.addSource(new ParalleSource());
        SingleOutputStreamOperator<Tuple1> mapTuple = source.map(line -> {
            return new Tuple1(line);
        }).returns(Types.TUPLE(Types.LONG));
        //Specifying keys via field positions is only valid for tuple data types. Type: Long
        //这里必须是tuple，然后按照tuple里面的哪个数据进行分组
        DataStream<Tuple1> longDataStream = mapTuple.partitionCustom(new Mypartition(), 0);
        //mapTuple.shuffle();
        //mapTuple.rescale();
        //mapTuple.rebalance();
        //mapTuple.broadcast();
        longDataStream.print().setParallelism(1);
        env.execute("start");
    }
}
